package com.marion.mrpc.server;

import com.marion.common.utils.ReflectUtils;
import com.marion.mrpc.Request;
import com.marion.mrpc.Response;
import com.marion.mrpc.codec.Decoder;
import com.marion.mrpc.codec.Encoder;
import com.marion.mrpc.transport.RequestHandler;
import com.marion.mrpc.transport.TransportServer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;

import java.io.InputStream;
import java.io.OutputStream;

/**
 * RPC服务端
 *
 * @author marion
 * @date 2021/9/4
 */
@Slf4j
public class RpcServer {

    private RpcServerConfig config;
    private TransportServer server;
    private Encoder encoder;
    private Decoder decoder;
    private ServiceManager serviceManager;
    private ServiceInvoker serviceInvoker;

    public RpcServer() {
        this(new RpcServerConfig());
    }

    public RpcServer(RpcServerConfig config) {
        this.config = config;
        this.server = ReflectUtils.newInstance(config.getTransport());
        this.server.init(config.getPort(), handler);

        this.encoder = ReflectUtils.newInstance(config.getEncoder());
        this.decoder = ReflectUtils.newInstance(config.getDecoder());

        this.serviceManager = new ServiceManager();
        this.serviceInvoker = new ServiceInvoker();
    }

    public void start() {
        this.server.start();
    }

    public void stop() {
        this.server.stop();
    }

    public <T> void register(Class<T> interfaceClass, T bean) {
        serviceManager.register(interfaceClass, bean);
    }

    /**
     * 处理http请求，加解码
     */
    private RequestHandler handler = new RequestHandler() {

        Response response = new Response();

        @Override
        public void onRequest(InputStream inputStream, OutputStream outputStream) {
            try {
                byte[] bytes = IOUtils.readFully(inputStream, inputStream.available());

                Request request = decoder.decode(bytes, Request.class);
                log.info("get request, {}", request);

                ServiceInstance service = serviceManager.lookup(request);
                log.info("get service, {}", service);
                Object invoke = serviceInvoker.invoke(service, request);
                response.setData(invoke);

            } catch (Exception e) {
                log.warn(e.getMessage(), e);
                response.setCode(1);
                response.setMessage("RpcServer get error: " + e.getClass().getName());
            } finally {
                byte[] encode = new byte[0];
                try {
                    encode = encoder.encode(response);
                    outputStream.write(encode);
                    log.info("RpcServer response");
                } catch (Exception e) {
                    log.warn("onRequest {}, {}", e.getMessage(), e);
                }
            }
        }
    };

}
